package czy.demo.config;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.hadoop.boot.properties.SpringHadoopProperties;
import org.springframework.data.hadoop.config.annotation.EnableHadoop;
import org.springframework.data.hadoop.config.annotation.SpringHadoopConfigs;
import org.springframework.data.hadoop.config.annotation.SpringHadoopConfigurerAdapter;
import org.springframework.data.hadoop.config.annotation.builders.HadoopConfigConfigurer;
import org.springframework.data.hadoop.config.annotation.builders.SpringHadoopConfigBuilder;
import org.springframework.data.hadoop.config.common.annotation.AnnotationBuilder;



/**
 * spring容器中hadoop的配置类
 */
@EnableHadoop
@Configuration
@EnableConfigurationProperties({ SpringHadoopProperties.class })
public class SpringHadoopConfig extends SpringHadoopConfigurerAdapter {

    @Autowired
    private SpringHadoopProperties shp;

    @Override
    public void configure(SpringHadoopConfigBuilder builder) throws Exception {
        super.configure(builder);
    }

    @Override
    public boolean isAssignable(AnnotationBuilder<SpringHadoopConfigs> builder) {
        return super.isAssignable(builder);
    }

    /**
     * 配置hadoop
     * @param config
     * @throws Exception
     */
    @Override
    public void configure(HadoopConfigConfigurer config) throws Exception {
        super.configure(config);
        /* hdfs文件系统url */
        config.fileSystemUri("hdfs://localhost:9000")
                .resourceManagerAddress(shp.getResourceManagerAddress())
                .jobHistoryAddress(shp.getJobHistoryAddress())
                .withProperties()
                .properties(shp.getConfig())
                .and()
                .withResources()
                .resources(shp.getResources())
                .and()
                .withSecurity()
                .namenodePrincipal(shp.getSecurity() != null ? shp.getSecurity().getNamenodePrincipal() : null)
                .rmManagerPrincipal(shp.getSecurity() != null ? shp.getSecurity().getRmManagerPrincipal() : null)
                .authMethod(shp.getSecurity() != null ? shp.getSecurity().getAuthMethod() : null)
                .userPrincipal(shp.getSecurity() != null ? shp.getSecurity().getUserPrincipal() : null)
                .userKeytab(shp.getSecurity() != null ? shp.getSecurity().getUserKeytab() : null);
    }

}
